import {
  Inject,
  Injectable,
  OnModuleDestroy,
  OnModuleInit,
} from "@nestjs/common";
import { Etcd3 } from "etcd3";
import { configuration } from "../../../config/config";
import { Observable, Observer } from "rxjs";

@Injectable()
export class EtcdService implements OnModuleInit, OnModuleDestroy {
  private client: Etcd3;
  private lease: any;
  private observer: Observer<any>;

  constructor(@Inject("SERVICE_NAME") private serviceName: string) {
    this.client = new Etcd3({
      hosts: configuration.etcdServer.host,
      auth: configuration.etcdServer.auth,
    });
    this.registerService(this.serviceName, {
      metadata: this.serviceName,
    });
  }

  async onModuleInit() {
    this.lease = this.client.lease(configuration.etcdServer.lease);
  }

  async registerService(serviceName, metadata) {
    const key = `${configuration.etcdServer.service_prefix}${serviceName}`;
    console.log(`注册新的etcd节点： ${key}`);
    const lease = this.client.lease(configuration.etcdServer.lease);
    await lease.put(key).value(JSON.stringify(metadata));
    lease.on("lost", async () => {
      await this.registerService(serviceName, metadata);
    });
  }

  async onModuleDestroy() {
    await this.lease.revoke();
  }

  async discoverService() {
    return await this.client
      .getAll()
      .prefix(configuration.etcdServer.service_prefix)
      .strings();
  }

  // 监听服务变更
  async watchService(): Promise<Observable<any>> {
    const watcher = await this.client
      .watch()
      .prefix(configuration.etcdServer.service_prefix)
      .create();
    watcher
      .on("put", async (event) => {
        console.log("新的服务节点添加:", event.key.toString());
        this.observer.next({
          systemName: event.key.toString(),
          status: true,
        });
      })
      .on("delete", async (event) => {
        console.log("服务节点删除:", event.key.toString());
        this.observer.next({
          systemName: event.key.toString(),
          status: false,
        });
      });
    return new Observable((observer: Observer<any>) => {
      this.observer = observer;
    });
  }
}
